package cn._51doit.flink.day04;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * 将两个数据流（流对应的数据类型，可以一样，也可以不一样）放到一起返回一个流，
 * 连接到一起后，两个流可以共享状态（共享数据）
 */
public class ConnectDemo1 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //5
        //6
        DataStreamSource<String> lines1 = env.socketTextStream("localhost", 8888);
        SingleOutputStreamOperator<Integer> nums1 = lines1.map(Integer::parseInt).setParallelism(3);

        DataStreamSource<String> lines2 = env.socketTextStream("localhost", 9999);
        SingleOutputStreamOperator<Long> nums2 = lines2.map(Long::parseLong);

        ConnectedStreams<Integer, Long> connectedStreams = nums1.connect(nums2);
        SingleOutputStreamOperator<String> res = connectedStreams.map(new CoMapFunction<Integer, Long, String>() {

            //定义共享的特殊变量（状态）

            //对第一个流进行map的方法
            @Override
            public String map1(Integer value) throws Exception {
                return value.toString();
            }

            //对第二个流进行map的方法
            @Override
            public String map2(Long value) throws Exception {
                return value.toString();
            }
        });

        res.print();

        env.execute();

    }
}
